package com.hkbigdata.processapi;

import com.hkbigdata.bean.WaterSensor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.IterationRuntimeContext;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @author liuanbo
 * @creat 2024-05-14-17:35
 * @see 2194550857@qq.com
 */
public class Flink03_ProcessJoinFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator = env.socketTextStream("localhost", 9999)
                .flatMap(new FlatMapFunction<String, WaterSensor>() {
                    @Override
                    public void flatMap(String value, Collector<WaterSensor> out) throws Exception {
                        String[] split = value.split(",");
                        out.collect(new WaterSensor(
                                split[0],
                                Long.valueOf(split[1]),
                                Double.valueOf(split[2])
                        ));
                    }
                });


        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator1 = env.socketTextStream("localhost", 8888)
                .flatMap(new FlatMapFunction<String, WaterSensor>() {
                    @Override
                    public void flatMap(String value, Collector<WaterSensor> out) throws Exception {
                        String[] split = value.split(",");
                        out.collect(new WaterSensor(
                                split[0],
                                Long.valueOf(split[1]),
                                Double.valueOf(split[2])
                        ));
                    }
                });

        //1.如果是接口，实现接口的话，那就需要重写接口里面所有方法（FlatMapFunction）
        //2.如果说是抽象类，继承抽象类，只要需要重写所有抽象方法（ProcessFunction）

        //第一层 ：匿名内部类 自定义类
        //第二层：KeyedProcessFunction 直接继承抽象类 AbstractRichFunction 包含父类所有方法+ontimer 定时器和output侧输出流+状态编程+获取当前key的getcurrentkey
        //第三层：AbstractRichFunction 直接实现接口 RichFunction，Serializable open() 生命周期方法和RuntimeContext运行时上下文访问。
        //第四层 ：RichFunction 继承 Function
        //第五层 ：Function 继承 Serializable
        /**
         * select
         * a.*,
         * b.*
         * from
         * (select id,ts,vc from table1) a
         * join
         * (select id,ts,vc from table1 ) b
         * on a.id=b.id;
         */
        waterSensorSingleOutputStreamOperator.
                join(waterSensorSingleOutputStreamOperator1)
                .where(WaterSensor::getId)
                .equalTo(WaterSensor::getId)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .apply(new JoinFunction<WaterSensor, WaterSensor, String>() {
                    @Override
                    public String join(WaterSensor first, WaterSensor second) throws Exception {
                        Double sum = first.getVc() + second.getVc();
                        return "first" + "id:" + first.getId() + "ts:" + first.getTs() + "sum:" + sum;
                    }
                }).print();

        env.execute();
    }
}
